package org.apache.rocketmq.client.java.impl.producer;


import com.google.common.base.Objects;
import com.google.common.collect.ImmutableList;
import com.google.common.hash.Hashing;
import com.google.common.math.IntMath;
import com.google.common.math.LongMath;
import org.apache.commons.lang3.RandomUtils;
import org.apache.rocketmq.client.java.misc.Utilities;
import org.apache.rocketmq.client.java.route.Broker;
import org.apache.rocketmq.client.java.route.Endpoints;
import org.apache.rocketmq.client.java.route.MessageQueueImpl;
import org.apache.rocketmq.client.java.route.TopicRouteData;

import javax.annotation.concurrent.Immutable;
import java.nio.charset.StandardCharsets;
import java.util.ArrayList;
import java.util.HashSet;
import java.util.List;
import java.util.Set;
import java.util.concurrent.atomic.AtomicInteger;
import java.util.function.Predicate;
import java.util.stream.Collectors;

/* @方法描述：发送消息时要用到的负载均衡器，这个负载均衡器会根据具体策略选择一个消息队列进行消息发送。
 */
@Immutable
public class PublishingLoadBalancer {

    // 当前索引 用于轮询策略
    private final AtomicInteger index;
    //存储所有可发送消息的消息队列的列表
    private ImmutableList<MessageQueueImpl> messageQueues;

    public PublishingLoadBalancer(TopicRouteData topicRouteData) {
        //初始化索引器索引，随机初始化是为了避免所有客户端同时选择同一个队列发送消息
        this.index = new AtomicInteger(RandomUtils.nextInt(0, Integer.MAX_VALUE));
        //只保留主节点且可写的节点
        final List<MessageQueueImpl> mqs = topicRouteData.getMessageQueues().stream()
                .filter(messageQueue -> messageQueue.getPermission().isWritable() && messageQueue.getBroker().getId() == Utilities.MASTER_BROKER_ID)
                .collect(Collectors.toList());

        //判断是否得到了可以使用的消息队列，没有则抛出异常
        if (mqs.isEmpty()) {
            throw new IllegalArgumentException("No writable message queue found, topiRouteData=" + topicRouteData);
        }
        this.messageQueues = ImmutableList.copyOf(mqs);
    }

    /**
     * @方法描述：根据消息分组选择消息队列的方法，在发送顺序消息时会调用该方法，把同一分组下的消息都发送到同一个消息队列中
     */
    public MessageQueueImpl takeMessageQueueByMessageGroup(String messageGroup) {
        //计算消息分组的哈希值
        final long hashCode = Hashing.sipHash24().hashBytes(messageGroup.getBytes(StandardCharsets.UTF_8)).asLong();
        //得到要选择的消息队列的索引
        final int index = LongMath.mod(hashCode, messageQueues.size());
        //选择消息队列，这个操作保证了，只要是同一分组下的消息，肯定会被路由到同一个消息队列中，方便消费者客户端顺序消费消息
        return messageQueues.get(index);
    }

    /*
     * @方法描述：选择多个消息队列发送消息的方法，这里大家可能会有疑惑，发送一条消息时为什么要选择多个消息队列呢？这是因为发送消息也有重试机制，发送失败的话就可以在选择好的消息队列进行轮询重试
     * 方法参数excluded是可能发生故障的Broker节点网络地址集合，如果选择的消息队列的broker节点网络地址在排除列表中，则不能选择该消息队列，count就是要选择的最大消息队列数量
     * 这个count其实就是请求发送失败的最大重试次数，能重试几次，就选择几个消息队列
     */
    public List<MessageQueueImpl> takeMessageQueues(Set<Endpoints> excluded, int count){
        //得到当前轮询索引
        int next = index.getAndIncrement();
        //创建候选集合，存放候选消息队列
        List<MessageQueueImpl> candidates = new ArrayList<>();
        //创建候选Broker集合，用于确保每个Broker只被选择一次
        Set<String> candidateBrokerNames = new HashSet<>();

        for (int i = 0; i < messageQueues.size(); i++) {
            //根据索引选择一个消息队列
            final MessageQueueImpl messageQueueImpl = messageQueues.get(IntMath.mod(next++, messageQueues.size()));
            //得到当前消息队列所属的Broker节点的信息，其实我们都知道，在5.0之后的版本中，Broker节点的地址已经被替换为Proxy服务端的地址了
            final Broker broker = messageQueueImpl.getBroker();
            //得到Broker节点名称
            final String brokerName = broker.getName();
            //判断当前选中的队列所属的Broker节点是否不在故障节点集合中，如果在故障节点集合中，则不能选择该队列
            //如果该Broker节点已经被选择过，则不能选择该队列
            if (!excluded.contains(broker.getEndpoints()) && !candidateBrokerNames.contains(brokerName)) {
                //把Broker名称添加到已选择集合，避免重复选择
                candidateBrokerNames.add(brokerName);
                //把队列添加到候选列表
                candidates.add(messageQueueImpl);
            }
            //如果选择的队列数量达到了目标数量，则返回候选列表
            if (candidates.size() >= count) {
                return candidates;
            }
        }

        //经过上面的循环之后，如果没有退出当前方法，则意味着没有找到足够可以使用的消息队列
        //那接下来就更改条件，重新寻找
        if (candidates.isEmpty()) {
            //再次遍历消息队列集合
            for (int i = 0; i < messageQueues.size(); i++) {
                //根据索引轮询得到消息队列
                final MessageQueueImpl messageQueueImpl = messageQueues.get(IntMath.mod(next++, messageQueues.size()));
                //得到消息队列所属的Broker信息
                final Broker broker = messageQueueImpl.getBroker();
                //得到Broker节点名称
                final String brokerName = broker.getName();
                //这次只判断是否没有重复选择Broker节点，如果没有重复选择，那就选中这个消息队列
                //可以看到，这里就不再判断Broker节点是否可用了，假如不可用，也可以被选中
                //这里算是一个衡量吧，反正发送消息时有重试机制，还有一点，可能这个故障节点此时已经恢复了
                if (!candidateBrokerNames.contains(brokerName)) {
                    //把Broker名称添加到已选择集合
                    candidateBrokerNames.add(brokerName);
                    //把队列添加到候选列表
                    candidates.add(messageQueueImpl);
                }
                //如果选择的队列数量达到了目标数量，则返回候选列表
                if (candidates.size() >= count) {
                    break;
                }
            }
        }
        //返回选择的队列列表
        return candidates;
    }

    @Override
    public boolean equals(Object o) {
        if (this == o) {
            return true;
        }
        if (o == null || getClass() != o.getClass()) {
            return false;
        }
        PublishingLoadBalancer that = (PublishingLoadBalancer) o;
        return Objects.equal(messageQueues, that.messageQueues);
    }

    @Override
    public int hashCode() {
        return Objects.hashCode(messageQueues);
    }
}